package org.korosoft.rusalad.syncnrun.tools;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;

/**
 * Semi-asynchronous task executor with limited queue length
 *
 * @author Dmitry Korotkov
 * @since 1.0
 */
public class BlockingQueueExecutor implements ExecutorService {
    private final BlockingQueue<Runnable> taskQueue;
    private final Collection<Thread> workers = new ArrayList<Thread>();
    private final CountDownLatch runningWorkers;
    private static final Runnable shutdownTask = new Runnable() {
        @Override
        public void run() {
        }
    };

    public BlockingQueueExecutor(String threadNamePrefix, int taskQueueSize, int threadCount) {
        runningWorkers = new CountDownLatch(threadCount);
        taskQueue = new ArrayBlockingQueue<Runnable>(taskQueueSize);
        for (int i = 1; i <= threadCount; i++) {
            Thread thread = new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        for (Runnable task = taskQueue.take(); task != shutdownTask; task = taskQueue.take()) {
                            task.run();
                        }
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            }, threadNamePrefix + i);
            workers.add(thread);
            thread.start();
        }
    }

    @Override
    public void shutdown() {
        for (int i = 0; i < workers.size(); i++) {
            try {
                taskQueue.put(shutdownTask);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    @Override
    public List<Runnable> shutdownNow() {
        for (Thread t : workers) {
            t.interrupt();
        }
        return new ArrayList<Runnable>(taskQueue);
    }

    @Override
    public boolean isShutdown() {
        return runningWorkers.getCount() == 0;
    }

    @Override
    public boolean isTerminated() {
        return runningWorkers.getCount() == 0;
    }

    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        return runningWorkers.await(timeout, unit);
    }

    @Override
    public <T> Future<T> submit(Callable<T> task) {
        throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public <T> Future<T> submit(Runnable task, T result) {
        throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public Future<?> submit(Runnable task) {
        throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
        throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
        throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
        throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public void execute(Runnable command) {
        try {
            taskQueue.put(command);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}
